package com.bwie.kafka.config;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @author 徐一杰
 * @date 2022/10/31 15:41
 * kafka消息发送回调
 */
@Component
public class KafkaSendResultHandler implements ProducerListener<Object, Object> {

    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    /**
     * bean 初始化方法 在 KafkaSendResultHandler bean 初始化的时候执行
     */
    @PostConstruct
    public void init() {
        this.kafkaTemplate.setProducerListener(this);
    }

    /**
     * 消息发送到 Kafka 成功的回调
     * @param producerRecord the actual sent record
     * @param recordMetadata the result of the successful send operation
     */
    @Override
    public void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata) {
        System.out.println("消息发送成功：" + producerRecord.toString());
    }

    /**
     * 息发送到 Kafka 失败的回调
     * @param producerRecord the failed record
     * @param recordMetadata The metadata for the record that was sent (i.e. the partition
     * and offset). If an error occurred, metadata will contain only valid topic and maybe
     * the partition. If the partition is not provided in the ProducerRecord and an error
     * occurs before partition is assigned, then the partition will be set to
     * RecordMetadata.UNKNOWN_PARTITION.
     * @param exception the exception thrown
     */
    @Override
    public void onError(ProducerRecord producerRecord, @Nullable RecordMetadata recordMetadata, Exception exception) {
        System.out.println("消息发送失败：" + producerRecord.toString() + exception.getMessage());
        // TODO 补偿  记录日志
    }

}
